package algorithms;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;

public class PageRankDriver extends Configured implements Tool {
	private static final Logger LOG = Logger.getLogger(PageRankDriver.class);
	
	public static int runIter = 0;
	private static final int MAX_ITERATIONS = 5;
	
	@Override
	public int run(String[] args) throws Exception {

		String inputPath = args[1] + "_tmp";
		String outputPath = args[1];
		int n = Integer.parseInt(args[2]);
		String nodes = args[3];

		LOG.info("Tool name: PageRank");
		LOG.info(" - inputDir: " + inputPath);
		LOG.info(" - outputDir: " + outputPath);
		LOG.info(" - src: " + n);
		LOG.info(" - totalNodes: " + nodes);

		Job job = new Job(getConf(), "PageRank");
		job.setJarByClass(PageRankDriver.class);

		job.setNumReduceTasks(n);

		job.getConfiguration().setInt("src", n);
		job.getConfiguration().set("total_nodes", nodes);
		job.getConfiguration().setInt("iteration", runIter);
		
		//FileInputFormat.addInputPath(job, new Path(inputPath));
		//FileOutputFormat.setOutputPath(job, new Path(outputPath));

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		job.setMapperClass(PageRankMapper.class);
		job.setReducerClass(PageRankReducer.class);
		
		if (args[4].equals("c")) {
			LOG.info("Changing partitioner.");
			job.setPartitionerClass(graphpartitioner.GraphPartitioner.class);
		}
		
		if (args.length > 5) {
			LOG.info("Using IMC.");
			job.getConfiguration().setBoolean("imc", true);
		}
		
		/* Basically, output a textfile at the end. Otherwise, output a sequence file */
		if (runIter > MAX_ITERATIONS + 2) {
			return 1;
		} else {
			//job.setOutputFormatClass(SequenceFileOutputFormat.class);
			
			FileInputFormat.addInputPath(job, new Path(outputPath + "/output-"
					+ (runIter)));
			FileOutputFormat.setOutputPath(job, new Path(outputPath + "/output-"
					+ (runIter + 1)));
		}
		
		// Delete the output directory if it exists already.
		//FileSystem.get(job.getConfiguration()).delete(new Path(outputPath),
		//		true);

		boolean result = job.waitForCompletion(true);

		return result ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new PageRankFormatter(), args);

		do {
			res = ToolRunner.run(new PageRankDriver(), args);
			runIter++;
		} while (res == 0);
		System.exit(res);
	}
}
